package com.krolik22.camel.learn;

import javax.jms.ConnectionFactory;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.camel.CamelContext;
import org.apache.camel.ExchangePattern;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.hawtdb.HawtDBAggregationRepository;
import org.apache.camel.component.jms.JmsComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class CamelFileWebJmsAggregateAndSplit {

	private static final Logger LOG = LoggerFactory.getLogger(CamelFileWebJmsAggregateAndSplit.class);

	private static final class HangupInterceptor extends Thread {
        Logger log = LoggerFactory.getLogger(this.getClass());
        CamelContext context;

        public HangupInterceptor(CamelContext context) {
        	this.context = context;
        }

        @Override
        public void run() {
            log.info("Received hang up - stopping the CamelFileWebJmsAggregateAndSplit instance.");
            try {
            	context.stop();
            } catch (Exception ex) {
                log.warn("Error during stopping the CamelFileWebJmsAggregateAndSplit instance.", ex);
            }
        }
    }
	
    public static void main(String args[]) throws Exception {
    	CamelFileWebJmsAggregateAndSplit instance = 
    			new CamelFileWebJmsAggregateAndSplit();
    	instance.execute(args);
    }
    
    public void execute(String[] args) throws Exception {
    	LOG.info("Doing init ...");
    	
        CamelContext context = new DefaultCamelContext();
        
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
        context.addComponent("my-jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        
        HangupInterceptor interceptor = 
        		new HangupInterceptor(context);
        Runtime.getRuntime().addShutdownHook(interceptor);
        
        /*AggregationRepository */
        final HawtDBAggregationRepository repo = 
        		new HawtDBAggregationRepository("repo1", "data/aggr/hawtdb.dat");
        repo.setBufferSize(16384);
        
    	LOG.info("Doing init ...done");
    	
    	LOG.info("Creating routes ...");
    	
    	// - FileToJMS
        context.addRoutes(new RouteBuilder() {
        	@Override
            public void configure() {
                from("file://inbox").
                 to("my-jms:rawIncomingData");
            }
        });
        
    	// - HTTPtoJMS
        context.addRoutes(new RouteBuilder() {
        	@Override
            public void configure() {
                from("jetty:http://0.0.0.0:8888/triggerprocessing").
                 setExchangePattern(ExchangePattern.InOnly).
                  to("my-jms:rawIncomingData").
                   transform().
                    constant("SUCCESS");
            }
        });
        
    	// - aggregate FileToJMS & HTTPtoJMS
    	// - JMStoLogger
        context.addRoutes(new RouteBuilder() {
        	@Override
            public void configure() {
                from("my-jms:rawIncomingData").
//                 aggregate(constant(Boolean.TRUE)).
//                 aggregationStrategy(new TriggerPropagationAggregationStrategy()).
                 aggregate(new TriggerPropagationAggregationStrategy()).
                  xpath("/item/@type", String.class).
                   aggregationRepository(repo).
                    eagerCheckCompletion().
                     completionPredicate(body().isEqualTo("GO")).
//                      transform(simple("Finally got : ${body}")).
                       to("my-jms:stagedIncomingDataMessage");
            }
        });
        
        // FIXME : DEBUG only
    	// - JMStoLogger
        context.addRoutes(new RouteBuilder() {
        	@Override
            public void configure() {
                from("my-jms:stagedIncomingData").
                 to("log:stagedIncomingDataMessage");
            }
        });
        
        // FIXME : DEBUG only
    	// - JMStoLogger
        context.addRoutes(new RouteBuilder() {
        	@Override
            public void configure() {
                from("my-jms:rawIncomingData").
                 to("log:rawIncomingData");
            }
        });
        
    	LOG.info("Creating routes ...done");
    	
    	LOG.info("Starting context ...done");
    	
        // Now everything is set up - lets start the context
        context.start();
        
    	LOG.info("Starting context ...done");
    }
    
    public void executeJmsToFile(String[] args) throws Exception {
        // START SNIPPET: e1
        CamelContext context = new DefaultCamelContext();
        // END SNIPPET: e1
        // Set up the ActiveMQ JMS Components
        // START SNIPPET: e2
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
        // Note we can explicit name the component
        context.addComponent("test-jms", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory));
        // END SNIPPET: e2
        // Add some configuration by hand ...
        // START SNIPPET: e3
        context.addRoutes(new RouteBuilder() {
            public void configure() {
                from("test-jms:queue:test.queue").to("file://test");
            }
        });
        // END SNIPPET: e3
        // Camel template - a handy class for kicking off exchanges
        // START SNIPPET: e4
        ProducerTemplate template = context.createProducerTemplate();
        // END SNIPPET: e4
        // Now everything is set up - lets start the context
        context.start();
        // Now send some test text to a component - for this case a JMS Queue
        // The text get converted to JMS messages - and sent to the Queue
        // test.queue
        // The file component is listening for messages from the Queue
        // test.queue, consumes
        // them and stores them to disk. The content of each file will be the
        // test we sent here.
        // The listener on the file component gets notified when new files are
        // found ... that's it!
        // START SNIPPET: e5
        for (int i = 0; i < 10; i++) {
            template.sendBody("test-jms:queue:test.queue", "Test Message: " + i);
        }
        // END SNIPPET: e5

        // wait a bit and then stop
        Thread.sleep(1000);
        context.stop();
    }
}
